package websocket

import (
	"context"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/gctx"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/os/gtimer"
	"github.com/gogf/gf/v2/text/gstr"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/gorilla/websocket"
	"sync/atomic"
	"time"
	"websocket/internal/consts"
)

type client struct {
	Closed     bool                       `json:"closed"`
	Unique     consts.WebsocketUniqueType `json:"unique"`
	Ipaddr     string                     `json:"ipaddr"`
	ConnTime   string                     `json:"conn_time"`
	Ctx        context.Context            `json:"-"`
	ctxClose   func()
	heartbeat  *gtimer.Entry
	writeCh    chan []byte
	WriteChNum int64 `json:"write_ch_num"` //用以待发送消息计数 atomic操作
	conn       *ghttp.WebSocket
}

func (c *client) read() {
	defer func() {
		g.Dump("read quit:" + gconv.String(c.Unique))
	}()
	for {
		select {
		case <-c.Ctx.Done():
			goto ERR
		default:
			if Manager().shutdown {
				//退出读协程
				return
			}
			if c.Closed {
				goto ERR
			}
			_, msg, err := c.conn.ReadMessage()
			if err != nil {
				goto ERR
			}
			if len(msg) == 0 {
				break
			}
			work := new(consts.WebsocketWorkContext)
			err = gjson.DecodeTo(msg, &work)
			if err != nil {
				goto ERR
			}
			work.Cmd = gstr.TrimAll(work.Cmd)
			work.Unique = c.Unique
			work.Original = msg
			if work.Cmd == "" {
				goto ERR
			}
			//client有消息上来重置心跳
			c.heartbeat.Reset()
			err = Manager().workPool.AddWithRecover(Context().InitLogger(ContextLogger{
				ClientTraceId: gctx.CtxId(c.Ctx),
				UpTime:        gtime.Now(),
				Original:      gconv.String(msg),
			}), func(ctx context.Context) {
				callBack := LogicManager().Call(ctx, work)
				if len(callBack) > 0 {
					err = Manager().WriteMsg(ctx, c.Unique, callBack)
					if err != nil {
						Manager().ClientCancel(ctx, c.Unique)
						return
					}
				}
			}, c.runRecover)
			if err != nil {
				goto ERR
			}
		}
	}
ERR:
	Manager().ClientCancel(c.Ctx, c.Unique)
	return
}

func (c *client) Pack(ctx context.Context, data interface{}) (encode []byte) {
	g.TryCatch(ctx, func(ctx context.Context) {
		encode = gjson.MustEncode(data)
	}, func(ctx context.Context, exception error) {
		encode = nil
	})
	return
}

//writeMsg 消息发送 所有消息均通过该方法发送 不要对writeCh投递消息
func (c *client) writeMsg(ctx context.Context, msg []*consts.WebsocketWorkRes) {
	if c.Closed == true {
		return
	}
	_ = g.Try(ctx, func(ctx context.Context) {
		c.writeCh <- c.Pack(ctx, msg)
		atomic.AddInt64(&c.WriteChNum, 1)
	})
}

func (c *client) write() {
	defer func() {
		g.Dump("write quit:" + gconv.String(c.Unique))
	}()
	for {
		select {
		case <-c.Ctx.Done():
			goto ERR
		case msg := <-c.writeCh:
			err := c.conn.WriteMessage(websocket.TextMessage, msg)
			atomic.AddInt64(&c.WriteChNum, -1)
			if err != nil {
				g.Log().Cat("write-err").Info(c.Ctx, msg)
				goto ERR
			}
		}
	}
ERR:
	Manager().ClientCancel(c.Ctx, c.Unique)
	return
}

func (c *client) Heartbeat() *gtimer.Entry {
	return gtimer.AddSingleton(c.Ctx, time.Second*30, func(ctx context.Context) {
		_ = g.Try(c.Ctx, func(ctx context.Context) {
			_ = Manager().WriteMsg(ctx, c.Unique, Manager().BuildRes(ctx, &consts.WebsocketWorkRes{
				Cmd:    consts.WebsocketCmdSysPing,
				Code:   0,
				Msg:    "ok",
				Unique: c.Unique,
				Data:   nil,
			}))
		})
	})
}

func (c *client) runRecover(ctx context.Context, err error) {
	g.Log().Cat("websocket").Info(ctx, Context().GetLogger(ctx), gerror.Stack(err))
}
